package cn.dmp.tools

import cn.dmp.utils.JedisPools
import org.apache.spark.sql.{Dataset, SparkSession}
import redis.clients.jedis.Jedis


//???

/**
  * 需求：媒体报表分析
  *     broadcast 广播变量的使用（将字典文件广播到Executor端）
  * 假设一个场景：如果字典文件一直在变，那Executor端如何获取该字典文件数据呢？（解决方案：如果Executor端要匹配Driver端的一直变化的字典文件数据，可以用redis数据库，将字典文件数据放到redis里面，以后需Executor要匹配字典数据的话可以直接读取redis数据库，并且redis读取的速度挺快的）
  * 广播变量的值能改变吗？一旦广播出去就不能改了。
  *
  * 本类目的：将字典文件存储到redis数据库！！！！！！！！
  */
object AppDictAnalyseToRedis {
  def main(args: Array[String]): Unit = {
    /*
    // 0.判断参数
    if (args.length != 1){
      println(
        """
          |cn.dmp.report.ProCityRpt_Version3
          |参数：
          |   dictFilePath
        """.stripMargin)
    }

    // 1.接收程序参数
    val Array(dictFilePath) = args

    // 2.创建SparkSession
    val sparkSession: SparkSession = SparkSession.builder().appName("ProCityRpt_Version3").master("local[*]")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") //spark优化：spark.serializer修改磁盘序列化方式
      .enableHiveSupport()
      .getOrCreate()

    // 将字典数据存储到redis
    /**
      * 注意：
      * 这里要将处理好的字段文件数据 存储到redis数据库里，用的 是jedis。
      * 我创建了一个执行类object名为JedisPoools。然后在该类里面获取jedis。：new JedisPool(new GenericObjectPoolConfig(), "192.168.161.11", 6379).getResource
      *
      */
    val dictDataSet: Dataset[String] = sparkSession.read.textFile(dictFilePath)
    dictDataSet.map(line => {
      val fields = line.split("\t", -1)
      (fields(4), fields(1))
    }).foreachPartition(itR => {
      val jedis: Jedis = JedisPools.getJedis()
      itR.foreach(t => {
        jedis.set(t._1, t._2)
      })

      jedis.close()
    })


    sparkSession.stop()

    //本地测试运行：（因为我没有那个dict字典数据文件，因此无法运行）!!!!!注意提前设置好参数：输入目录和输出目录
    /**
      * 输入目录：E:\大数据培训视频\项目+spark面试+flink+项目\项目\DMP\字典数据。。。。这个数据我没有！！
      */
  */
  }
}



